Improve event-driven reporter with cycle detection - #31
Conversation
There was a problem hiding this comment.
I like these changes, but want to change setRPCClient(...) to give the concrete HTTP client used for subscriptions the same fallback that you get when normal RPC fallback switches endpoints. (feel free to pull in the changes from my branch here or I can merge it all in tomorrow after manual test on palmito.
| grpcConn *grpc.ClientConn | ||
| grpcClient daemontypes.GrpcClient | ||
|
|
||
| rpcClient *rpchttp.HTTP // direct reference for WebSocket subscriptions |
There was a problem hiding this comment.
Adding this import in a new branch pr-31-review
| defer wg.Done() | ||
|
|
||
| // Try to use event-driven detection (reacts within milliseconds of each new block). | ||
| // Falls back to 200ms ticker polling if the WebSocket subscription fails. |
There was a problem hiding this comment.
added fallback behavior to next listed rpc in .env in this branch https://github.com/tellor-io/layer-daemons/tree/pr-31-review
…ine start (PR tellor-io#31 review)
0xSpuddy
left a comment
0xSpuddy
left a comment
There was a problem hiding this comment.
it's just import ordering, so I will approve the pr and merge.
| // Channel closed; try to re-subscribe on a fallback RPC endpoint. | ||
| unsubscribe() | ||
| c.logger.Warn("block subscription channel closed, trying fallback RPC endpoint") | ||
| blockCh, unsubscribe, err = c.subscribeNewBlocks(ctx) |
There was a problem hiding this comment.
suggest using subscribeNewBlocksFallback instead to make consistent with existing fallback behavior
blockCh, unsubscribe, err = c.subscribeNewBlocksFallback(ctx, fmt.Errorf("subscription channel closed"))
There was a problem hiding this comment.
There is a separate resource cleanup issue that we can solve after this is merged, or as part of this work: rpcEndpointManager creates fresh RPC clients, and the WebSocket path starts them with httpClient.Start(), but the cleanup only unsubscribes and never calls Stop(). When a subscription is replaced, a fallback endpoint is tried, Subscribe fails after Start(), or the primary RPC health check starts a client and then rejects it, the old CometBFT client workers/connections appear to be left running. Can we make the owner of each started RPC client stop it when it is no longer used?
| @@ -14,6 +14,7 @@ import ( | |||
|
|
|||
2 participants
Summary
Replaces the 200ms polling loop in
MonitorCyclelistQuerywith a CometBFTNewBlockWebSocket subscription. The reporter now reacts to each new block within ~10-50ms instead of waiting up to 200ms, giving more of the 3.1s cycle window for price fetching, tx building, and broadcasting.Changes
rpcClient (*rpchttp.HTTP)directly onClientto access theSubscribeAPIMonitorCyclelistQuerysubscribes totm.event='NewBlock'on startupmonitorCyclelistQueryPolling()Expected Impact
Reduces cycle detection latency by ~150ms on average. Combined with the existing 15s broadcast timeout, this is expected to bring steady-state miss rates from ~11% closer to the 2-4% range.